package com.patsnap.data.npd.dw.etl.utils;

import com.alibaba.fastjson.JSON;
import com.alibaba.fastjson.JSONObject;
import com.alibaba.fastjson.TypeReference;
import org.apache.hadoop.conf.Configuration;
import org.apache.spark.api.java.function.MapFunction;
import org.apache.spark.sql.Dataset;
import org.apache.spark.sql.Row;
import org.apache.spark.sql.RowFactory;
import org.apache.spark.sql.SparkSession;
import org.apache.spark.sql.catalyst.encoders.ExpressionEncoder;
import org.apache.spark.sql.catalyst.encoders.RowEncoder;
import org.apache.spark.sql.types.DataType;
import org.apache.spark.sql.types.DataTypes;
import org.apache.spark.sql.types.StructField;
import org.apache.spark.sql.types.StructType;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import scala.collection.mutable.WrappedArray;

import java.io.*;
import java.net.HttpURLConnection;
import java.net.URL;
import java.util.*;
import java.util.stream.Collectors;

public class EtlUtils {

    private final static Logger logger = LoggerFactory.getLogger(EtlUtils.class);

    public static void addHadoopDefaultResource(Configuration configuration) {
        addHadoopClassPathResource(configuration, "/core-site.xml");
        addHadoopClassPathResource(configuration, "/hdfs-site.xml");
    }

    public static void addHadoopClassPathResource(Configuration configuration, String resourceClasspath) {
        InputStream is = EtlUtils.class.getResourceAsStream(resourceClasspath);
        configuration.addResource(is);
    }


    public static Object formatField(int index , Row row , DataType dataType,int logId) {
        if(null == row.get(index)){
            return "";
        }
        Object object;
        switch (dataType.typeName()){
            case "struct":
                Map<String, Object> rowMapMessage = new HashMap<>();
                Row row0 = row.getStruct(index);
                StructField[] structFields = row0.schema().fields();
                for (int i = 0; i < structFields.length; i++) {
                    DataType fieldDataType = structFields[i].dataType();
                    Object o = formatField(i,row0,fieldDataType,logId);
                    rowMapMessage.put(structFields[i].name(), o);
                }
                object = rowMapMessage;
                break;
            case "array":
                List list = row.getList(index);
                List list1 = new ArrayList(list.size());
                list.forEach(e->{
                    list1.add(formatArray(e, logId));
                });
                object = list1;
                break;
            case "timestamp":
                if(logId == -10000 || logId == 178){
                    object = row.get(index);
                }else{
                    object = row.get(index).toString();
                }
                break;
            default:
                object = row.get(index);
        }
        return object;
    }

    private static Object formatArray(Object ob , Integer logId){
        if(ob instanceof Row){
            Row row1 = (Row)ob;
            Map<String, Object> rowMapMessage0 = new HashMap<>();
            StructField[] structFields0 = row1.schema().fields();
            for (int i = 0; i < structFields0.length; i++) {
                DataType fieldDataType = structFields0[i].dataType();
                if(null != row1.get(i)){
                    rowMapMessage0.put(structFields0[i].name(), EtlUtils.formatField(i,row1,fieldDataType,logId));
                }else{
                    rowMapMessage0.put(structFields0[i].name(), new String());
                }
            }
            return rowMapMessage0;
        }else if(ob instanceof WrappedArray){
            WrappedArray wrappedArray = (WrappedArray<Object>) ob;
            int len = wrappedArray.length();
            List list = new ArrayList(len);
            for (int i = 0; i < len ; i++){
                list.add(formatArray(wrappedArray.apply(i),logId));
            }
            return list;
        }else{
            return ob;
        }
    }

    /**
     * build StructType from field expression
     *
     * @param fieldTypes [service:String,timestamp:Timestamp,env:Byte]
     * @return
     */
    public static StructType buildStructType(String[] fieldTypes) {
        List<StructField> structFields = Arrays.stream(fieldTypes).map(x -> {
            String[] fieldPair = x.split(":");
            String fieldName = fieldPair[0];
            String fieldType = fieldPair[1];
            DataType dataType; // = dataTypeMap.get(fieldType.toLowerCase());
            switch (fieldType.toLowerCase().trim()) {
                case "binary":
                    dataType = DataTypes.BinaryType;
                    break;
                case "boolean":
                    dataType = DataTypes.BooleanType;
                    break;
                case "byte":
                    dataType = DataTypes.ByteType;
                    break;
                case "calendarinterval":
                    dataType = DataTypes.CalendarIntervalType;
                    break;
                case "date":
                    dataType = DataTypes.DateType;
                    break;
                case "double":
                    dataType = DataTypes.DoubleType;
                    break;
                case "float":
                    dataType = DataTypes.FloatType;
                    break;
                case "int":
                case "integer":
                    dataType = DataTypes.IntegerType;
                    break;
                case "long":
                    dataType = DataTypes.LongType;
                    break;
                case "null":
                    dataType = DataTypes.NullType;
                    break;
                case "short":
                    dataType = DataTypes.ShortType;
                    break;
                case "string":
                    dataType = DataTypes.StringType;
                    break;
                case "timestamp":
                    dataType = DataTypes.TimestampType;
                    break;
                case "array":
                    dataType = DataTypes.createArrayType(DataTypes.StringType);
                    break;
                default:
                    throw new UnsupportedOperationException("unsupport data type:" + fieldType);
            }
            return DataTypes.createStructField(fieldName, dataType, true);
        }).collect(Collectors.toList());

        return DataTypes.createStructType(structFields);
    }

    public static ExpressionEncoder<Row> createRowExpressionEncoder(String[] fieldTypes) {
        StructType structType = EtlUtils.buildStructType(fieldTypes);
        return RowEncoder.apply(structType);
    }

    public static List<String> jsonParseArray(String jsonStr) {
        return JSON.parseArray(jsonStr, String.class);
    }

    public static String getErrorMessage(Throwable throwable) {
        if (throwable == null) {
            return "";
        }

        StringWriter sw = null;
        PrintWriter pw = null;
        try {
            sw = new StringWriter();
            pw = new PrintWriter(sw);
            throwable.printStackTrace(pw);
            pw.flush();
            sw.flush();
        } catch (Exception ex) {
            ex.printStackTrace();
        } finally {
            if (sw != null) {
                try {
                    sw.close();
                } catch (IOException e1) {
                    e1.printStackTrace();
                }
            }
            if (pw != null) {
                pw.close();
            }
        }

        return (throwable.getMessage() == null ? "" : throwable.getMessage()) + "\n" + sw.toString();
    }

    public static String httpGet(String address) {
        HttpURLConnection conn;
        URL url;
        BufferedReader reader = null;
        StringBuilder result = new StringBuilder();
        try {
            url = new URL(address);
            conn = (HttpURLConnection) url.openConnection();
            conn.setConnectTimeout(10000);
            conn.connect();

            InputStream in = conn.getInputStream();
            reader = new BufferedReader(new InputStreamReader(in));
            String s = "";
            while ((s = reader.readLine()) != null) {
                result.append(s);
            }
            System.out.println("url:" + address + "=>" + result);
            reader.close();
            in.close();
            conn.disconnect();
        } catch (Exception e) {
            throw new RuntimeException("http get error, url:" + address, e);
        }

        return result.toString();
    }

    public static String httpPostJson(String address, String json) {
        String result = "";
        BufferedReader reader = null;
        try {
            URL url = new URL(address);
            HttpURLConnection conn = (HttpURLConnection) url.openConnection();
            conn.setRequestMethod("POST");
            conn.setDoOutput(true);
            conn.setDoInput(true);
            conn.setUseCaches(false);
            conn.setRequestProperty("Connection", "Keep-Alive");
            conn.setRequestProperty("Charset", "UTF-8");
            // 设置文件类型:
            conn.setRequestProperty("Content-Type", "application/json; charset=UTF-8");
            // 设置接收类型否则返回415错误
            //conn.setRequestProperty("accept","*/*")此处为暴力方法设置接受所有类型，以此来防范返回415;
            conn.setRequestProperty("accept", "application/json");
            // 往服务器里面发送数据
            if (json != null && !"".equals(json.trim())) {
                byte[] jsonBytes = json.getBytes();
                // 设置文件长度
                conn.setRequestProperty("Content-Length", String.valueOf(jsonBytes.length));
                OutputStream out = conn.getOutputStream();
                out.write(jsonBytes);
                out.flush();
                out.close();
            }

            if (conn.getResponseCode() == 200) {
                reader = new BufferedReader(new InputStreamReader(conn.getInputStream()));
                result = reader.readLine();
            } else if (conn.getResponseCode() >= 300) {
                throw new RuntimeException("http response code " + conn.getResponseCode());
            }
        } catch (Exception e) {
            throw new RuntimeException("http post error, url:" + address, e);
        } finally {
            if (reader != null) {
                try {
                    reader.close();
                } catch (IOException e) {
                    e.printStackTrace();
                }
            }
        }
        return result;
    }

    public static Dataset<Row> formatSchema (Dataset<Row> dataFrame, SparkSession sparkSession){
        StructType structType = dataFrame.schema();
        StructField[] structFields = structType.fields();
        List filedLists = new ArrayList<String>();
        for (StructField structField : structFields) {
            String fieldDataType = structField.dataType().typeName();
            String filedName = structField.name();
            sparkSession.log().warn("fieldName:{"+filedName+"},fieldType:{"+fieldDataType+"}");
            if(fieldDataType.equals("struct") || fieldDataType.equals("array")){
                filedLists.add(filedName + ":string");
                sparkSession.log().warn("struct || array will be cast string,fieldName:{"+filedName+"},fieldType:{"+fieldDataType+"}");
            }else{
                filedLists.add(filedName + ":" + fieldDataType);
            }
        }
        String[] fieldTypes = (String[]) filedLists.toArray(new String[filedLists.size()]);
        ExpressionEncoder<Row> rowExpressionEncoder = EtlUtils.createRowExpressionEncoder(fieldTypes);
        logger.warn("fieldTypes is:{}",fieldTypes.toString());

        Dataset<Row> df = dataFrame.map(new MapFunction<Row, Row>() {
            @Override
            public Row call(Row row) throws Exception {
                Object[] objects = new Object[structFields.length];
                for (int i = 0; i < structFields.length; i++) {
                    DataType fieldDataType = structFields[i].dataType();
                    if(null != row.get(i)){
                        if(fieldDataType.typeName().equals("struct") || fieldDataType.typeName().equals("array")){
                            Object value =  EtlUtils.formatField(i,row,fieldDataType,-10000);
                            String formatValue = JSONObject.toJSONString(value);
                            objects[i] = formatValue;
                        }else{
                            objects[i] =  row.get(i);
                        }
                    }else{
                        if(fieldDataType.typeName().equals("struct") || fieldDataType.typeName().equals("array") || fieldDataType.typeName().equals("string")){
                            objects[i] = new String();
                        }else{
                            objects[i] = row.get(i);
                        }

                    }
                }

                return RowFactory.create(objects);
            }
        },rowExpressionEncoder);
        return df;
    }

    public static String toJSONString(Object o) {
        return JSON.toJSONString(o);
    }

    public static byte[] toJSONBytes(Object o) {
        return JSON.toJSONBytes(o);
    }

    public static <T> T parseObject(String jsonStr, Class<T> clazz) {
        return JSON.parseObject(jsonStr, clazz);
    }

    public static <T> T parseObject(String text,
                                    TypeReference<T> type) {
        return JSON.parseObject(text, type);
    }

    public static JSONObject parseObject(String text) {
        return JSON.parseObject(text);
    }

    public static void main(String[] args) {
    }
}
